package cn.z2huo.demo.kafka.apache.consumer;

import cn.z2huo.demo.kafka.apache.constant.KafkaConfigConstant;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * <p>
 *
 * @author z2huo
 */
public class ConsumerProperties {

    public static Properties getConsumerProperties(){

        // 创建 Kafka 消费者配置
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfigConstant.BOOTSTARP_SERVERS);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 当消费的消费者组没有消费过时，从头开始消费
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 服务器在没有足够数据立即满足 fetch.min.bytes 所要求的情况下，在响应提取请求之前会阻塞的最长时间
        properties.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "1000");
        // 服务器应为获取请求返回的最小数据量。如果可用数据量不足，请求将在回答请求之前等待足够的数据累积。
        properties.setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1");
        // 消费者是否自动提交偏移量
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 消费者偏移量自动提交到 Kafka 的频率
        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        return properties;
    }

}
